package cn.mesmile.flink.demo;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * @author zb
 * @date 2021/8/21 11:43
 * @Description  本地UI 监控
 *     nc -l -p 8888
 *     - win | linux 需要安装
 *       - win 百度搜索博文参考不同系统安装
 *     - 下载地址 [https://eternallybored.org/misc/netcat/](https://eternallybored.org/misc/netcat/)
 *     - linux 安装
 *      - yum install -y netcat
 *      - yum install -y nc
 */
public class FlinkSourceLocalUIEnvApp {

    public static void main(String[] args) throws Exception {
        // import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
        // import org.apache.flink.configuration.Configuration;
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        //env.setParallelism(1);
        /*
         * STREAMING 流处理
         * BATCH 批处理
         * AUTOMATIC 根据source类型自动选择运行模式
         */
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
        // 从 socket 中监听数据
        DataStream<String> stringDataStream = env.socketTextStream("127.0.0.1",8888);
        DataStream<String> flatMapDataStream = stringDataStream.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public void flatMap(String value, Collector<String> out) throws Exception {
                String[] arr = value.split(",");
                for (String word : arr) {
                    out.collect(word);
                }
            }
        });
        flatMapDataStream.print("结果");

        //DataStream需要调用execute,可以取个名称
        env.execute("data stream job");
    }
}
